package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Demo19Checkpoint {

  def main(args: Array[String]): Unit = {


    //环境配置对象
    val conf = new SparkConf()

    //指定任务名
    conf.setAppName("pi")
    conf.setMaster("local")


    //创建spark 环境,式写spark代码的入口
    val sc = new SparkContext(conf)

    //设置checkpoint路径
    sc.setCheckpointDir("data/checkpoint")


    val studentRDD: RDD[String] = sc.textFile("data/students.txt")

    val kvRDD: RDD[(String, Int)] = studentRDD.map(line => {
      println("map")

      val split: Array[String] = line.split(",")

      (split(4), split(2).toInt)
    })


    /**
      * checkpoint流程
      * 1、当第一个job执行完成之后会从最后一个rdd向前回溯，对做了checkpoint的rdd打上标记
      * 2、另启动一个新的job重新计算rdd的数据，并将rdd的数据保存到hdfs中
      * 3、后续如果在由job用到了这个rdd就可以直接使用hdfs中的数据了
      *
      *
      * 优化;可以checkpoint之前将rdd的数据缓存
      *
      */

    kvRDD.cache()

    kvRDD.checkpoint()


    //统计班级的人数
    kvRDD
      .map(kv => (kv._1, 1))
      .reduceByKey(_ + _)
      .foreach(println)


    //统计班级的平均年龄
    kvRDD
      .groupByKey()
      .map(kv => {
        val clazz: String = kv._1
        val ages: List[Int] = kv._2.toList
        val avgAge: Double = ages.sum.toDouble / ages.size
        (clazz, avgAge)
      })
      .foreach(println)


    kvRDD
      .groupByKey()
      .map(kv => {
        val clazz: String = kv._1
        val ages: List[Int] = kv._2.toList
        val avgAge: Double = ages.sum.toDouble / ages.size
        (clazz, avgAge)
      })
      .foreach(println)


    while (true) {

    }

  }

}
